package com.java.springcloud.util;

import com.java.springcloud.config.ZookeeperProperties;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.zookeeper.CreateMode;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author dushougudu
 */
@Data
@Slf4j
public class ZkClient {
    private CuratorFramework curatorFramework;

    public ZkClient(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
    }

    /**
     * 初始化zookeeper客户端
     */
    public void init(ZookeeperProperties zookeeperProperties) {
//        try {
//            RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperProperties.getElapsedTimeMs(),
//                    zookeeperProperties.getRetryCount());
//            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
//                    .connectString(zookeeperProperties.getAddress()).retryPolicy(retryPolicy)
//                    .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
//                    .connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
//                    .namespace("mongo");
//            if (StringUtils.isNotEmpty(zookeeperProperties.getDigest())) {
//                builder.authorization("digest", zookeeperProperties.getDigest().getBytes("UTF-8"));
//                builder.aclProvider(new ACLProvider() {
//                    @Override
//                    public List<ACL> getDefaultAcl() {
//                        return ZooDefs.Ids.CREATOR_ALL_ACL;
//                    }
//
//                    @Override
//                    public List<ACL> getAclForPath(final String path) {
//                        return ZooDefs.Ids.CREATOR_ALL_ACL;
//                    }
//                });
//            }
//            curatorFramework = builder.build();
//            curatorFramework.start();
//
//            initLocalCache("/test");
//
//            curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
//                @Override
//                public void stateChanged(CuratorFramework curatorFramework, ConnectionState state) {
//                    if (state == ConnectionState.LOST) {
//                        //连接丢失
//                        log.info("lost session with zookeeper");
//                    } else if (state == ConnectionState.CONNECTED) {
//                        //连接新建
//                        log.info("connected with zookeeper");
//                    } else if (state == ConnectionState.RECONNECTED) {
//                        log.info("reconnected with zookeeper");
//                    }
//                }
//            });
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
    }

    /**
     * 初始化本地缓存
     *
     * @param watchRootPath
     * @throws Exception
     */
    private void initLocalCache(String watchRootPath) throws Exception {
//        cache = CuratorCache.builder(curatorFramework, watchRootPath).build();
//        CuratorCacheListener listener = CuratorCacheListener
//                .builder()
//                .forAll(new CuratorCacheListener() {
//                    @Override
//                    public void event(Type type, ChildData oldData, ChildData data) {
//                        log.info("event:" + type.name());
//                        log.info("老节点：" + oldData.toString());
//                        log.info("新节点：" + data.toString());
//                    }
//                })
//                .build();
//        cache.listenable().addListener(listener);
//        cache.start();
    }


    public void stop() {
        curatorFramework.close();
    }


    /**
     * 创建节点
     *
     * @param mode     节点类型
     *                 1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
     *                 2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
     *                 3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
     *                 4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
     * @param path     节点名称
     * @param nodeData 节点数据
     */
    public void createNode(CreateMode mode, String path, String nodeData) {
        try {
            //使用creatingParentContainersIfNeeded()之后Curator能够自动递归创建所有所需的父节点
            curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes("UTF-8"));
        } catch (Exception e) {
            log.error("注册出错", e);
        }
    }

    /**
     * 创建节点
     *
     * @param mode 节点类型
     *             1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
     *             2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
     *             3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
     *             4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
     * @param path 节点名称
     */
    public void createNode(CreateMode mode, String path) {
        try {
            //使用creatingParentContainersIfNeeded()之后Curator能够自动递归创建所有所需的父节点
            curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
        } catch (Exception e) {
            log.error("注册出错", e);
        }
    }

    /**
     * 删除节点数据
     *
     * @param path
     */
    public void deleteNode(final String path) {
        try {
            deleteNode(path, true);
        } catch (Exception ex) {
            log.error("{}", ex);
        }
    }


    /**
     * 删除节点数据
     *
     * @param path
     * @param deleteChildre 是否删除子节点
     */
    public void deleteNode(final String path, Boolean deleteChildre) {
        try {
            if (deleteChildre) {
                //guaranteed()删除一个节点，强制保证删除,
                // 只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到删除节点成功
                curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
            } else {
                curatorFramework.delete().guaranteed().forPath(path);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 设置指定节点的数据
     *
     * @param path
     * @param datas
     */
    public void setNodeData(String path, byte[] datas) {
        try {
            curatorFramework.setData().forPath(path, datas);
        } catch (Exception ex) {
            log.error("{}", ex);
        }
    }

    /**
     * 获取指定节点的数据
     *
     * @param path
     * @return
     */
    public byte[] getNodeData(String path) {
        try {
            curatorFramework.getData().forPath(path);
            return curatorFramework.getData().forPath(path);
        } catch (Exception ex) {
            log.error("{}", ex);
        }
        return null;
    }

    /**
     * 获取数据时先同步
     *
     * @param path
     * @return
     */
    public byte[] synNodeData(String path) {
        curatorFramework.sync();
        return getNodeData(path);
    }

    /**
     * 判断路径是否存在
     *
     * @param path
     * @return
     */
    public boolean isExistNode(final String path) {
        curatorFramework.sync();
        try {
            return null != curatorFramework.checkExists().forPath(path);
        } catch (Exception ex) {
            return false;
        }
    }


    /**
     * 获取节点的子节点
     *
     * @param path
     * @return
     */
    public List<String> getChildren(String path) {
        List<String> childrenList = new ArrayList<>();
        try {
            childrenList = curatorFramework.getChildren().forPath(path);
        } catch (Exception e) {
            log.error("获取子节点出错", e);
        }
        return childrenList;
    }


    /**
     * 获取读写锁
     *
     * @param path
     * @return
     */
    public InterProcessReadWriteLock getReadWriteLock(String path) {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(curatorFramework, path);
        return readWriteLock;
    }

    /**
     * 在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理
     */
    Executor pool = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());

    /**
     * 监听数据节点的变化情况
     *
     * @param watchPath
     * @param listener
     */
    public void watchPath(String watchPath, CuratorCacheListener listener) {
        CuratorCache cache = CuratorCache.builder(curatorFramework, watchPath).build();
        cache.listenable().addListener(listener, pool);
        cache.start();
    }

}
